package orange.com.openstack.sahara.api;

import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import net.minidev.json.JSONObject;
import orange.com.openstack.sahara.data.Job;
import orange.com.openstack.sahara.httpConnection.RestCall;
import orange.com.openstack.tools.StringConverter;

import com.google.gson.Gson;
import com.jayway.jsonpath.JsonPath;

/**
 * class allowing the interaction with the Jobs Rest service
 * 
 * @author osvaldo.cocucci@gmail.com
 * 
 */
/**
 * @author mfrh5781
 * 
 */
public class Jobs {
	String endpoint;
	String tenantId;
	String token;
	Gson gson;

	/**
	 * @param endpoint
	 * @param tenantId
	 * @param token
	 * 
	 *            constructor
	 */
	public Jobs(String endpoint, String tenantId, String token) {
		gson = new Gson();
		this.endpoint = endpoint;
		this.tenantId = tenantId;
		this.token = token;
	}

	/**
	 * 
	 * This operation returns the list of all created Jobs. This operation does
	 * not require a request body.
	 */
	public List<Job> listJobs() {
		List<Job> jobs = new ArrayList<Job>();
		RestCall caller = new RestCall();
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs", null, "GET", token);
		try {
			if (connection.getResponseCode() != 200)
				return null;
			String JSONreponse = caller.getResponse(connection);
			connection.disconnect();
			List<JSONObject> jobsJSON = JsonPath.read(JSONreponse, "$.jobs[*]");

			for (JSONObject json : jobsJSON) {
				jobs.add(gson.fromJson(json.toJSONString(), Job.class));
			}
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return jobs;
	}

	/**
	 * @param dataSourceId
	 *            This operation returns the information about the specified
	 *            Job. This operation does not require a request body.
	 * @return
	 */
	public Job showJobs(String jobId) {
		RestCall caller = new RestCall();
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs/" + jobId, null, "GET", token);
		try {
			if (connection.getResponseCode() != 200)
				return null;
			String JSONreponse = caller.getResponse(connection);
			connection.disconnect();
			return gson.fromJson(((JSONObject) JsonPath.read(JSONreponse,
					"$.job")).toJSONString(), Job.class);
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * @param desc
	 * @param main
	 *            e.g "84248975-3c82-4206-a58d-6e7fb3a563fd"
	 * @param libs
	 *            e.g "508fc62d-1d58-4412-b603-bdab307bb926"
	 * @param type
	 *            can be "pig"
	 * @param name
	 *            can be "pig-job-example"
	 * 
	 *            This operation shows information about the created Job object.
	 */
	public Job createJob(String desc, String[] main, String[] libs,
			String type, String name) {
		RestCall caller = new RestCall();
		String payload = "{\"description\": \"" + desc + "\",\"mains\": "
				+ Arrays.toString(main) + ",\"libs\":" + Arrays.toString(libs)
				+ ",\"type\": \"" + type + "\",\"name\": \"" + name + "\"}";
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs", payload, "POST", token);

		try {
			if (connection.getResponseCode() != 202)
				return null;
			String JSONreponse = caller.getResponse(connection);
			connection.disconnect();
			return gson.fromJson(((JSONObject) JsonPath.read(JSONreponse,
					"$.job")).toJSONString(), Job.class);
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * @param jobId
	 *            Removes the Job object This operation returns nothing. This
	 *            operation does not require a request body.
	 * @return
	 */
	public boolean deleteJob(String jobId) {
		RestCall caller = new RestCall();
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs/" + jobId, null, "DELETE", token);
		try {
			if (connection.getResponseCode() == 204)
				return true;
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * @param jobType
	 *            This operation returns hints for configuration parameters
	 *            which can be applied during job execution. This operation does
	 *            not require a request body.he id
	 */
	public void getJobConfigHint(String jobType) {
		RestCall caller = new RestCall();
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs/config-hints/" + jobType, null, "GET", token);
		try {
			connection.connect();
			System.out.println(caller.getResponse(connection));
			connection.disconnect();
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * @param clusterId
	 * @param jobId
	 * @param args
	 * @param username
	 * @param pwd
	 * @param main_class
	 * @return the id of the runing job that different job types support
	 *         different combinations of configs, args, and params. The Elastic
	 *         Data Processing (EDP) document discusses these differences.
	 */
	public String executeJob(String clusterId, String jobId, String args[],
			String username, String pwd, String main_class) {
		RestCall caller = new RestCall();
		String payload = "{\"cluster_id\": \""
				+ clusterId
				+ "\",\"job_configs\": {\"configs\": {\"fs.swift.service.sahara.username\": \""
				+ username + "\",\"fs.swift.service.sahara.password\": \""
				+ pwd + "\",\"edp.java.main_class\": \"" + main_class
				+ "\"},\"args\": " + StringConverter.Array2String(args) + "}}";

		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs/" + jobId + "/execute", payload, "POST", token);
		try {
			connection.connect();
			String JSONreponse = caller.getResponse(connection);

			connection.disconnect();
			return JsonPath.read(JSONreponse, "$.job_execution.id");

		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}


	/**
	 * 
	 * This operation returns the list of all created Jobs. This operation does
	 * not require a request body.
	 */
	public boolean exist(String jobName) {
		RestCall caller = new RestCall();
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs", null, "GET", token);
		try {
			connection.connect();

			String JSONreponse = caller.getResponse(connection);
			List<Object> jobs = JsonPath.read(JSONreponse,
					"$.jobs[?(@.name == '" + jobName + "')]");
			connection.disconnect();
			if (jobs.size() > 0) {
				return true;
			}
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * 
	 * This operation returns the list of all created Jobs. This operation does
	 * not require a request body.
	 */
	public String getId(String jobName) {
		RestCall caller = new RestCall();
		HttpURLConnection connection = caller.connect(endpoint + tenantId
				+ "/jobs", null, "GET", token);
		try {
			connection.connect();

			String JSONreponse = caller.getResponse(connection);
			List<Object> binarie = JsonPath.read(JSONreponse,
					"$.jobs[?(@.name == '" + jobName + "')].id");
			connection.disconnect();
			return (String) binarie.get(0);
		} catch (MalformedURLException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

}
